<?php

/**
 * Адаптер доставки по средством с помощью модуля nginx:
 * https://github.com/wandenberg/nginx-push-stream-module
 * User: s.utkin
 * Date: 29.08.14
 * Time: 15:51
 */
namespace NotificationDelivery\Delivery;

use NotificationDelivery\Delivery;
use Pinenut\Model2\Finder;
use Users\Queue;

class PushStream extends Delivery {
	static $onlineTime = 60;
	static $justBeenHereTime = 600;
	static $channelLifeTime = 86400;

	static $configClassName = "PushStreamConfig";
	
	/**
	 * Модель состояния канала
	 * 
	 * @var PushStreamState
	 */
	protected $stateModel;
	protected $messages = [ ];
	protected function getStateModel($create = false) {
		if (! $this->stateModel) {
			// ищем модель
			$this->stateModel = PushStreamState::find ()->where ( 'subscriber', $this->subscriber )->one ();
		}
		
		if (! $this->stateModel && $create) {
			// создаём модель
			$this->stateModel = new PushStreamState ();
			$this->stateModel->subscriber = $this->subscriber;
			$this->stateModel->channel = getmypid () . uniqid ();
		}
		
		if ( $this->stateModel && $create) {
			// если создали или нашли модель
			// переводим её статус в онлайн и уведомляем
			if ($this->stateModel->state != PushStreamState::ONLINE) {
				Queue::enqueue ( [
						'type' => \UsersNotificationDeliveryConfig::TYPE_USER_ONLINE_STATUS_CHANGED,
						'data' => [
								'subscriber' => ( string ) $this->stateModel->subscriber,
								'state' => PushStreamState::ONLINE,
						] 
				] );
				$this->stateModel->set('state', PushStreamState::ONLINE);
				$this->stateModel->set('onlineTime', time ());
				$this->stateModel->save ();
			}
		}
		return $this->stateModel;
	}
	/**
	 * Возвращаем адрес канала
	 *
	 * @return string
	 */
	public function subscribe() {
		// ищем запись о подписчике,
		// если не нашли - создаем новый
		$state = $this->getStateModel ( true );
		
		// генерим адрес
		$sub = static::getUrl('sub');
		if (!empty ($sub)) {
			return $sub . $state->channel;
		}
	}
	
	public function addMessage($message) {
		$this->messages [] = $message;
	}
	public function publish($message = null) {
		// добавляем сообщение
		if (null !== $message) {
			$this->addMessage ( $message );
		}
		
		// по подписчику получаем канал
		$state = $this->getStateModel ();
		
		// если канал онлайн, отправляем туда сообщение
		if ($state && ($state->state != PushStreamState::OFFLINE)) {
			static::pub ( $state->channel, $this->messages );
		}
	}
	
	/**
	 *
	 * @param string $channel        	
	 * @param string $messages        	
	 */
	protected static function pub($channel, $messages) {
		$curl = curl_init ();
		
		curl_setopt ( $curl, CURLOPT_RETURNTRANSFER, true );
		curl_setopt ( $curl, CURLOPT_POST, true );
		curl_setopt ( $curl, CURLOPT_HTTPHEADER, array (
				'Content-Type: text/html' 
		) );
		curl_setopt ( $curl, CURLOPT_URL, static::getUrl('pub') . $channel );
		
		$results = array ();
		if (! is_array ( $messages )) {
			$messages = array (
					$messages
			);
		}
		
		foreach ( $messages as $message ) {
			curl_setopt ( $curl, CURLOPT_POSTFIELDS, $message . ',' );
			$results [] = curl_exec ( $curl );
		}
		
		curl_close ( $curl );
	}
	
	/**
	 * Возвращаем онлайн
	 * 
	 * @return boolean null
	 */
	public function state() {
		
		$model = PushStreamState::find ()
			->where ( 'subscriber', $this->subscriber )
			->fields(['state'])
			->active(false)
			->one ();

		if (isset($model['state'])) {
			return $model['state'];
		}
	}
	
	/**
	 * Возвращаем время, когда пользователь был онлайн
	 * 
	 *  @return int $onlineTime
	 */
	public function getOnlineLastActivity() {
		$model = PushStreamState::find ()
			->where ('subscriber', $this->subscriber )
			->fields(['onlineTime'])
			->active(false)
			->one();
		
		if (isset($model['onlineTime'])) {
			return (int)$model['onlineTime'];
		}
	}
	
	/**
	 * Устанавливаем стату в онлайн, если канал с подписчиками
	 * @param unknown $channels
	 */
	static function setOnline($channels) {
		// ищем записи по каналам
		$states = PushStreamState::find ()
			->where ( 'channel', $channels, Finder::OP_IN )
			->active ( false )
			->all ();
		
		$ids = array_column ( (array)$states, '_id' );
		//print_r($ids);
		if (count($ids)) {				

// 			$fe = fopen('php://stderr', 'w');
// 			fwrite($fe, print_r($ids, true));
				
			// обновляем
			PushStreamState::update ()
				->set ( 'state', PushStreamState::ONLINE)
				->set ( 'onlineTime', time ())
				->where ( '_id', $ids, Finder::OP_IN )
				->execute ();
	
			// генерим события о изменении состояний с оффлайн на онлайн
			foreach ($states as $state) {
				if ($state['state'] != PushStreamState::ONLINE) {
					echo "\nto Online: ".$state['subscriber'];
					Queue::enqueue ( [ 
							'type' => \UsersNotificationDeliveryConfig::TYPE_USER_ONLINE_STATUS_CHANGED,
							'data' => [ 
									'subscriber' => ( string ) $state['subscriber'],
									'state' => PushStreamState::ONLINE,
							] 
					] );
				}
			}
		}
	}
	
	/**
	 * Протухших онлайнеров низвергаем в только что бывших
	 * За одно генерим эвенты
	 */
	static function setJustBeen() {
		$states = PushStreamState::find ()
			->where ( 'state', PushStreamState::ONLINE)
			->where ( 'onlineTime', time() - static::$onlineTime, Finder::OP_LT)
			->active ( false )
			->all ();
		
		
		$ids = array_column ( (array)$states, '_id' );
		
		if (count($ids)) {
			// обновляем
			PushStreamState::update ()
				->set ( 'state', PushStreamState::JUSTBEEN)
				->where ( '_id', $ids )
				->execute ();
				
				// генерим события о изменении состояний с оффлайн на онлайн
			foreach ( $states as $state ) {
				if ($state ['state'] != PushStreamState::JUSTBEEN) {
					echo "\nto JBH: ".$state['subscriber'];
					Queue::enqueue ( [ 
							'type' => \UsersNotificationDeliveryConfig::TYPE_USER_ONLINE_STATUS_CHANGED,
							'data' => [ 
									'subscriber' => ( string ) $state['subscriber'],
									'state' => PushStreamState::JUSTBEEN 
							] 
					] );
				}
			}
		}
	}
	
	/**
	 * Протухших онлайнеров низвергаем в оффлайн
	 * За одно генерим эвенты
	 */
	static function setOffline() {
		$states = PushStreamState::find ()
			->where ( 'state', PushStreamState::JUSTBEEN )
			->where ( 'onlineTime', time () - static::$justBeenHereTime, Finder::OP_LT )
			->active ( false )->all ();
		
		$ids = array_column ( ( array ) $states, '_id' );
		
		if (count ( $ids )) {
			// обновляем
			PushStreamState::update ()
				->set ( 'state', PushStreamState::OFFLINE )
				->where ( '_id', $ids )
				->execute ();
			
			// генерим события о изменении состояний с оффлайн на онлайн
			foreach ( $states as $state ) {
				if ($state ['state'] != PushStreamState::OFFLINE) {
					echo "\nto Offline: " . $state ['subscriber'];
					Queue::enqueue ( [ 
							'type' => \UsersNotificationDeliveryConfig::TYPE_USER_ONLINE_STATUS_CHANGED,
							'data' => [ 
									'subscriber' => ( string ) $state ['subscriber'],
									'state' => PushStreamState::OFFLINE
							]
					] );
				}
			}
		}
	}
	
	static function getOnlinersFromNginx() {

		// Получаем статистика
		$curl = curl_init();	
		curl_setopt($curl, CURLOPT_RETURNTRANSFER, true);
		curl_setopt($curl, CURLOPT_URL, static::getUrl('stat'));
		$result = curl_exec($curl);
		curl_close($curl);
		
		$stat = json_decode($result, true);
		// обновляем онлайнеров
		// ищем записи, модифицируем
		// отправляем сообщения на переход в онлайн
		$onliners = [];
		foreach ($stat['infos'] as $channel) {
			if ($channel['subscribers'] > 0) {
				$onliners[] = $channel['channel'];
			}
		}

		return $onliners;
		
	}
	
	/**
	 * Удаляем все протухшие каналы
	 */
	static function clear() {
		PushStreamState::delete()
			->where ( 'onlineTime', time () - static::$channelLifeTime, Finder::OP_LT )
			->execute();
	}
	
	static function getUrl($type) {
		$class = static::$configClassName;
		return $class::get($type);
	}
} 